package cn.lei.netty.herostory.mq;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 消息队列生产者
 *
 * @author lei
 */
public class MqProducer {
    /**
     * 日志对象
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(MqProducer.class);
    /**
     * 消息队列生产者
     */
    private static DefaultMQProducer producer = null;

    /**
     * 私有化类默认构造器
     */
    private MqProducer() {
    }

    /**
     * 初始化
     */
    public static void init() {
        try {
            //创建消息生产者
            DefaultMQProducer newProducer = new DefaultMQProducer("herostory");
            //指定nameServer地址
            newProducer.setNamesrvAddr("127.0.0.1:9876");
            newProducer.start();
            newProducer.setRetryTimesWhenSendAsyncFailed(3);

            producer = newProducer;

            LOGGER.info("消息队列 ( 生产者 ) 连接成功!");
        } catch (MQClientException ex) {
            // 记录错误日志
            LOGGER.error(ex.getMessage(), ex);
        }
    }

    /**
     * 发送消息
     *
     * @param topic 主题
     * @param msg   消息对象
     */
    public static void sendMsg(String topic, Object msg) {
        if (topic == null || msg == null) {
            return;
        }

        Message newMsg = new Message();
        newMsg.setTopic(topic);
        newMsg.setBody(JSONObject.toJSONBytes(msg));

        //发送消息
        try {
            producer.send(newMsg);
        } catch (Exception ex) {
            // 记录错误日志
            LOGGER.error(ex.getMessage(), ex);
        }
    }
}
